/*
 * db_mgr.cpp
 *
 *  Created on: Jan 9, 2014
 *      Author: killerg
 */

#include "db_mgr.h"
#include "share/util.h"

static void error_callback(lcb_t conn, lcb_error_t error, const char* errinfo);
static void config_callback(lcb_t conn, lcb_configuration_t config);
static void get_callback(lcb_t conn, const void* cookie, lcb_error_t error, const lcb_get_resp_t* resp);
static void store_callback(lcb_t conn, const void* cookie, lcb_storage_t operation, lcb_error_t error, const lcb_store_resp_t* resp);
static void view_callback(lcb_http_request_t request, lcb_t conn, const void* cookie, lcb_error_t error, const lcb_http_resp_t* resp);
static void remove_callback(lcb_t conn, const void *cookie, lcb_error_t error, const lcb_remove_resp_t *resp);
static lcb_io_opt_t create_libevent_io_ops(struct event_base *base);
static db_error_t trans_error(lcb_error_t err);

bool db_mgr_t::init(
        const std::string &host,
        const std::string &user,
        const std::string &passwd,
        const std::string &bucket)
{

    this->host = host;
    this->user = user;
    this->passwd = passwd;
    this->bucket = bucket;

    base = event_base_new();
    if (NULL == base)
        return false;

    ioops = create_libevent_io_ops(base);
    if (NULL == ioops)
        return false;

    finished = false;
    _state = Invalid;

    do {
        update();
    } while (Initing == _state);

    return _state == Running;
}

void db_mgr_t::destroy()
{
    if (_state != Invalid) {
    	finished = true;
    	_state = Destroying;
    	do update();
    	while (_state != Invalid);
    }

    if (ioops) {
        lcb_destroy_io_ops(ioops);
        ioops = NULL;
    }

    if (base) {
        event_base_free(base);
        base = NULL;
    }
}

void db_mgr_t::update()
{
	switch (_state) {
	case Invalid:
		if (!finished) {
		    assert(!has_pending());
		    create_instance();
		}
		break;
	case Destroying:
		if (!has_pending()) {
			destroy_instance();
		}
		break;
	}

    event_base_loop(base, EVLOOP_NONBLOCK);
}

bool db_mgr_t::repair(int try_num)
{
	if (try_num <= 0)
		try_num = 999999999;

	do update();
	while (!ok() && --try_num > 0);

	return ok();
}

void db_mgr_t::wait()
{
    lcb_wait(_inst);
}

bool db_mgr_t::gets(const std::vector<std::string> &keys, cb_ejb_t cb)
{
    int len = keys.size();
    if (len <= 0)
        return true;

	assert(++operating == 1);

    auto cookie = new cookie_t;
    cookie->remain = len;
    cookie->cb_ejb = cb;
    auto pcmds = new lcb_get_cmd_t*[len];
    auto cmds = new lcb_get_cmd_t[len];
    for (int i=0; i<len; ++i) {
        pcmds[i] = &cmds[i];
        pcmds[i]->v.v0.key = keys[i].c_str();
        pcmds[i]->v.v0.nkey = keys[i].length();
    }

    lcb_error_t err = lcb_get(_inst, cookie, cookie->remain, pcmds);
    if (err == LCB_SUCCESS) {
    	++pending_gets;
    } else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::gets Failed to setup get request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

    delete[] pcmds;
    delete[] cmds;

	assert(--operating == 0);

    return err == LCB_SUCCESS;
}

bool db_mgr_t::get(const char* key, cb_ej_t cb)
{
	assert(++operating == 1);

    lcb_get_cmd_t cmd(key);
    const lcb_get_cmd_t *cmds[1] = {&cmd};

    auto cookie = new cookie_t;
    cookie->cb_ej = cb;

    lcb_error_t err = lcb_get(_inst, cookie, 1, cmds);
    if (err == LCB_SUCCESS) {
        ++pending_get;
    } else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::get Failed to setup get request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

	assert(--operating == 0);

	return err == LCB_SUCCESS;
}

bool db_mgr_t::dels(const std::vector<std::string> &keys, cb_eb_t cb)
{
	int len = keys.size();
    if (len <= 0)
        return true;

	assert(++operating == 1);

	auto cookie = new cookie_t;
    cookie->remain = len;
    cookie->cb_el = cb;
    auto pcmds = new lcb_remove_cmd_t*[len];
    auto cmds = new lcb_remove_cmd_t[len];
    for (int i=0; i<len; ++i) {
        pcmds[i] = &cmds[i];
        pcmds[i]->v.v0.key = keys[i].c_str();
        pcmds[i]->v.v0.nkey = keys[i].length();
    }

    lcb_error_t err = lcb_remove(_inst, cookie, cookie->remain, pcmds);
    if (err != LCB_SUCCESS) {
    	++pending_dels;
    } else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::dels Failed to setup remove request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

    delete[] pcmds;
    delete[] cmds;

	assert(--operating == 0);

	return err == LCB_SUCCESS;
}

bool db_mgr_t::del(const char* key, cb_e_t cb)
{
	assert(++operating == 1);

    lcb_remove_cmd_t cmd(key);
    const lcb_remove_cmd_t *cmds[1] = {&cmd};

    auto cookie = new cookie_t;
    cookie->cb_e = cb;

    lcb_error_t err = lcb_remove(_inst, cookie, 1, cmds);
    if (err == LCB_SUCCESS) {
        ++pending_del;
    } else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::del Failed to setup remove request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

	assert(--operating == 0);

    return err == LCB_SUCCESS;
}

bool db_mgr_t::_store(const char* key, const Json::Value &value, cb_e_t cb, lcb_storage_t store_type, time_t expire)
{
	assert(++operating == 1);

    std::string str = to_str(value);
    lcb_store_cmd_t cmd(store_type, key, strlen(key), str.c_str(), str.length(), 0, expire);
    const lcb_store_cmd_t *cmds[1] = {&cmd};

    auto cookie = new cookie_t;
    cookie->cb_e = cb;

    lcb_error_t err = lcb_store(_inst, cookie, 1, cmds);
    if (err == LCB_SUCCESS) {
        if (store_type == LCB_SET)
        	++pending_set;
        else
        	++pending_add;
	} else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::_store Failed to set up store request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

    assert(--operating == 0);

    return err == LCB_SUCCESS;
}


bool db_mgr_t::view(const char* path, cb_ej_t cb)
{
	assert(++operating == 1);

    lcb_http_request_t req;
    lcb_http_cmd_t cmd(path, strlen(path), NULL, 0, LCB_HTTP_METHOD_GET, 0, "application/json");

    auto cookie = new cookie_t;
    cookie->cb_ej = cb;

    lcb_error_t err = lcb_make_http_request(_inst, cookie, LCB_HTTP_TYPE_VIEW, &cmd, &req);
    if (err == LCB_SUCCESS) {
        ++pending_view;
	} else {
    	_state = Destroying;
        UTIL_LOG(E,
                "db_mgr_t2::view Failed to set up view request: %s\n",
                lcb_strerror(_inst, err));
        delete cookie;
    }

    assert(--operating == 0);

    return err == LCB_SUCCESS;
}

void db_mgr_t::destroy_instance()
{
	assert(_state == Destroying);

	if (_inst != NULL) {
		lcb_destroy(_inst);
		_inst = NULL;
	}
	_state = Invalid;
}

void db_mgr_t::create_instance()
{
	assert(_state == Invalid && _inst == NULL);

    lcb_t inst;
    lcb_error_t error;
    lcb_create_st copts(host.c_str(), user.c_str(), passwd.c_str(), bucket.c_str(), ioops);
    lcb_error_t err = lcb_create(&inst, &copts);
    if (err != LCB_SUCCESS) {
        UTIL_LOG(E, "Failed to create a libcouchbase instance: %s\n", lcb_strerror(NULL, error));
        exit(EXIT_FAILURE);
    }

    /* Set up the callbacks */
    lcb_set_error_callback(inst, error_callback);
    lcb_set_configuration_callback(inst, config_callback);
    lcb_set_get_callback(inst, get_callback);
    lcb_set_store_callback(inst, store_callback);
    lcb_set_http_complete_callback(inst, view_callback);
    lcb_set_remove_callback(inst, remove_callback);
    lcb_set_timeout(inst, 60*1000*1000);
    lcb_set_cookie(inst, this);

    err = lcb_connect(inst);
    if (err != LCB_SUCCESS) {
    	UTIL_LOG(E, "Failed to connect libcouchbase instance: %s\n", lcb_strerror(NULL, error));
        lcb_destroy(inst);
        return;
    }

    ++pending_conf;
    _inst = inst;
    _state = Initing;
}


static void error_callback(lcb_t conn, lcb_error_t error, const char *errinfo)
{
    db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(conn);
    assert(db);

    db->_state = db_mgr_t::Destroying;

    UTIL_LOG(E, "%s %s error_callback ERROR: %s %s\n",
            db->host.c_str(),
            db->bucket.c_str(),
            lcb_strerror(conn, error),
            errinfo);

}

static void config_callback(lcb_t inst, lcb_configuration_t config)
{
    db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(inst);
    assert(db);

    db->_state = db_mgr_t::Running;
    db->pending_conf = 0;

    UTIL_LOG(I, "couchbase conn created %s:%s", db->host.c_str(), db->bucket.c_str());
}


static void get_callback(lcb_t conn, const void *cookie, lcb_error_t error, const lcb_get_resp_t *resp)
{
	db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(conn);
	assert(db);
	assert(db->_state == db_mgr_t::Running || db->_state == db_mgr_t::Destroying);

	if (resp->v.v0.nbytes > db->max_objsize)
		db->max_objsize = resp->v.v0.nbytes;
	if (resp->v.v0.nbytes < db->min_objsize)
		db->min_objsize = resp->v.v0.nbytes;

    auto cook = (db_mgr_t::cookie_t*)cookie;
    if (NULL != cook) {
        if (cook->remain < 0) {
        	if (db)
        		--db->pending_get;

            if (cook->cb_ej) {
                std::string str((char*)resp->v.v0.bytes, resp->v.v0.nbytes);
                Json::Value json = to_json(str);
                if (json.isNull())
                	error = LCB_KEY_ENOENT;
                cook->cb_ej(trans_error(error), json);
            }
            delete cook;
        } else {
            --cook->remain;
            if (cook->cb_ejb) {
                std::string str((char*)resp->v.v0.bytes, resp->v.v0.nbytes);
                Json::Value json = to_json(str);
                if (json.isNull())
                	error = LCB_KEY_ENOENT;
                cook->cb_ejb(trans_error(error), json, cook->remain == 0);
            }
            if (cook->remain == 0) {
            	if (db)
            		--db->pending_gets;
                delete cook;
            }
        }
    } else {
        UTIL_LOG(E, "get_callback ERROR: %s\n", lcb_strerror(conn, error));
    }
}

static void store_callback(lcb_t conn, const void *cookie, lcb_storage_t operation, lcb_error_t error, const lcb_store_resp_t *resp)
{
	db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(conn);
	assert(db);
	assert(db->_state == db_mgr_t::Running || db->_state == db_mgr_t::Destroying);

	if (operation == LCB_SET)
		--db->pending_set;
	else
		--db->pending_add;

	auto cook = (db_mgr_t::cookie_t*)cookie;
    if (NULL != cook) {
        if (cook->cb_e)
            cook->cb_e(trans_error(error));
        delete cook;
    } else {
        UTIL_LOG(E, "store_callback ERROR: %s\n", lcb_strerror(conn, error));
    }
}

static void remove_callback(lcb_t conn, const void *cookie, lcb_error_t error, const lcb_remove_resp_t *resp)
{
	db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(conn);
	assert(db);
	assert(db->_state == db_mgr_t::Running || db->_state == db_mgr_t::Destroying);

    auto cook = (db_mgr_t::cookie_t*)cookie;
    if (NULL != cook) {
        if (cook->remain < 0) {
            --db->pending_del;
            if (cook->cb_e) {
                cook->cb_e(trans_error(error));
            }
            delete cook;
        } else {
            if (cook->cb_el) {
                cook->cb_el(trans_error(error), cook->remain == 1);
                --cook->remain;
            }
            if (cook->remain == 0) {
                --db->pending_dels;
                delete cook;
            }
        }
    } else {
        UTIL_LOG(E, "remove_callback ERROR: %s\n", lcb_strerror(conn, error));
    }
}

static void view_callback( lcb_http_request_t request, lcb_t conn, const void *cookie, lcb_error_t error, const lcb_http_resp_t *resp)
{
	db_mgr_t* db = (db_mgr_t*)lcb_get_cookie(conn);
	assert(db);
	assert(db->_state == db_mgr_t::Running || db->_state == db_mgr_t::Destroying);

	--db->pending_view;
	if (resp->v.v0.nbytes > db->max_objsize)
		db->max_objsize = resp->v.v0.nbytes;
	if (resp->v.v0.nbytes < db->min_objsize)
		db->min_objsize = resp->v.v0.nbytes;

	auto cook = (db_mgr_t::cookie_t*) cookie;

    if (NULL != cook) {
        if (cook->cb_ej) {
            std::string str((char*)resp->v.v0.bytes, resp->v.v0.nbytes);
            Json::Value json = to_json(str);
            cook->cb_ej(trans_error(error), json);
        }
        delete cook;
    } else {
        UTIL_LOG(E, "view_callback ERROR: %s\n", lcb_strerror(0, error));
    }
}

static db_error_t trans_error(lcb_error_t err) {
	switch (err) {
    case LCB_SUCCESS:
        return DBE_Success;
    case LCB_KEY_ENOENT:
        return DBE_NoExist;
    case LCB_KEY_EEXISTS:
        return DBE_Exist;
    case LCB_AUTH_ERROR:
    case LCB_DELTA_BADVAL:
    case LCB_E2BIG:
    case LCB_EINVAL:
    case LCB_ERANGE:
    case LCB_NOT_STORED:
    case LCB_NOT_SUPPORTED:
    case LCB_UNKNOWN_COMMAND:
    case LCB_UNKNOWN_HOST:
    case LCB_INVALID_CHAR:
    case LCB_DUPLICATE_COMMANDS:
        return DBE_ArgsErr;
    case LCB_ETIMEDOUT:
    	return DBE_TryLater;
    case LCB_EBUSY:
    case LCB_ENOMEM:
    case LCB_ETMPFAIL:
    case LCB_CLIENT_ETMPFAIL:
    case LCB_BUSY:
    case LCB_NETWORK_ERROR:
    case LCB_CONNECT_ERROR:
    case LCB_EINTERNAL:
        return DBE_TryMoreLater;
    case LCB_DLOPEN_FAILED:
    case LCB_DLSYM_FAILED:
    case LCB_NOT_MY_VBUCKET:
    case LCB_PROTOCOL_ERROR:
    case LCB_BUCKET_ENOENT:
    case LCB_CLIENT_ENOMEM:
    case LCB_EBADHANDLE:
    case LCB_SERVER_BUG:
    case LCB_PLUGIN_VERSION_MISMATCH:
    case LCB_INVALID_HOST_FORMAT:
    case LCB_DURABILITY_ETOOMANY:
    case LCB_NO_MATCHING_SERVER:
    case LCB_BAD_ENVIRONMENT:
    case LCB_INVALID_USERNAME:
    case LCB_CONFIG_CACHE_INVALID:
    case LCB_SASLMECH_UNAVAILABLE:
    case LCB_TOO_MANY_REDIRECTS:
    default:
        return DBE_CanNotRepair;
    }
}

static lcb_io_opt_t create_libevent_io_ops(struct event_base *base)
{
    struct lcb_create_io_ops_st ciops;
    lcb_io_opt_t ioops;
    lcb_error_t error;

    memset(&ciops, 0, sizeof(ciops));
    ciops.v.v0.type = LCB_IO_OPS_LIBEVENT;
    ciops.v.v0.cookie = base;

    error = lcb_create_io_ops(&ioops, &ciops);
    if (error != LCB_SUCCESS) {
        UTIL_LOG(E, "Failed to create an IOOPS structure for libevent: %s\n",
                lcb_strerror(NULL, error));
        exit(EXIT_FAILURE);
    }

    return ioops;
}

const char* db_strerror(db_error_t err) {
	switch (err) {
	case DBE_Success:
		return "DBE_Success";
	case DBE_Exist:
		return "DBE_Exist";
	case DBE_NoExist:
		return "DBE_NoExist";
	case DBE_ArgsErr:
		return "DBE_ArgsErr";
	case DBE_TryLater:
		return "DBE_TryLater";
	case DBE_TryMoreLater:
		return "DBE_TryMoreLater";
	case DBE_CanNotRepair:
		return "DBE_CanNotRepair";
	default:
		return "db unknow err";
	}
}
